MapReduce ETL开发案例

基础数据

1
baidu	CN	A	E	[17/Jul/2018:17:07:50 +0800]	2	223.104.18.110	-	112.29.213.35:80	0	v2.go2yd.com	GET	http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c4b96b2226fd3f4c.mp4_bd.mp4	HTTP/1.1	-	bytes 13869056-13885439/25136186	TCP_HIT/206	112.29.213.35	video/mp4	17168	16384	-:0	0	0	-	-	-	11451601	-	"JSP3/2.0.14"	"-"	"-"	"-"	http	-	2	v1.go2yd.com	0.002	25136186	16384	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	-	1531818470104-11451601-112.29.213.66#2705261172	644514568

项目新建

  • idea新建项目

  • 选择自己安装的maven

  • 删除app文件

pom.xml文件修改

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

<!-- 配置版本-->
  <properties>
    <hadoop.version>2.6.0-cdh5.7.0</hadoop.version>
  </properties>

- 这里的版本名称，不一定一定要跟线上一样
- maven打包分两种，胖包和瘦包，我们一般采用瘦包的打包方式

<!--添加远程仓库地址-->
    <repositories>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>

<!--添加junit依赖和hadoop依赖-->
    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
            <scope>test</scope>
        </dependency>

        <!--添加hadoop依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
    </dependencies>

代码开发

LogUtils功能类开发

  • 代码

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    package com.ruozedata.hadoop.utils;

    import java.text.DateFormat;
    import java.text.ParseException;
    import java.text.SimpleDateFormat;
    import java.util.Locale;

    public class LogUtils {
        DateFormat sourceFormat = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss", Locale.ENGLISH);
        DateFormat targetFormat = new SimpleDateFormat("yyyyMMddHHmmss");

        /**
         * 日志文件解析，对内容进行字段的处理
         * 按/t分隔
         */
        public String parse(String log){

            String result = "";
            String[] split = log.split("\t");
            String cdn = split[0];
            String region = split[1];
            String level = split[3];
            String timeStr = split[4];
            String time = timeStr.substring(1,timeStr.length()-7);
            try {
                time = targetFormat.format(sourceFormat.parse(time));
                String ip = split[6];
                String domain = split[10];
                String url = split[12];
                String trafic = split[20];

                StringBuilder builder = new StringBuilder();
                builder.append(cdn).append("\t")
                        .append(region).append("\t")
                        .append(level).append("\t")
                        .append(time).append("\t")
                        .append(ip).append("\t")
                        .append(domain).append("\t")
                        .append(url).append("\t")
                        .append(trafic).append("\t");
                result = builder.toString();
            } catch (ParseException e) {
                e.printStackTrace();
            }
            return result;
        }
    }

  • 解释

    1
    2
    3
    4
    5
    6
    7
    在功能类里我们做了这么几步操作
    - 通过parse接收一个log参数
    - 对log参数按照'\t'进行分割
    - 按照数组的下标依次取出我们需要的字段
    - 对时间字段进行了处理
      - 将[17/Jul/2018:17:07:50 +0800]格式的时间转化为20180717170750
    - 将字段进行重新按照'\t'分隔，进行组装return
LogETLMapper类开发
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
package com.ruozedata.hadoop.mapreduce.mapper;

import com.ruozedata.hadoop.utils.LogUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class LogETLMapper extends Mapper<LongWritable,Text,NullWritable,Text>{

    /**
     * 通过MapReduce框架的map方法进行数据清洗
     * 进来一条数据就按照我们的规则进行解析
     * @param key
     * @param value
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        int lenth = value.toString().split("\t").length;
        if (lenth==72){
            LogUtils utils = new LogUtils();
            String results = utils.parse(value.toString());

            if(StringUtils.isNotBlank(results)){
                context.write(NullWritable.get(),new Text(results));
            }
        }else {

        }
    }
}
  • 解释
    1
    2
    3
    4
    5
    6
    7
    - 在LogETLMapper中首先我们继承了Mapper类
      - 由于ETL是不需要reduce操作的，这里我们可以看到Mapper定义的reduce参数的key是NullWritable
    - 重写map方法
    - 判断value的lenth是否是我们预期的72个字符
      - 如果不是，我们丢掉这条数据
      - 如果是，调用我们开发好的LogUtils.parse方法对数据进行处理
    - 调用context.write方法将数据写入上下文
LogETLDriver类开发
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
package com.ruozedata.hadoop.mapreduce.driver;

import com.ruozedata.hadoop.mapreduce.mapper.LogETLMapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class LogETLDriver {
    public static void main(String[] args) throws Exception {
        if (args.length != 2){
            System.err.println("please input 2 params: input output");
            System.exit(0);
        }
        String input = args[0];
        String output = args[1];//"output/d=20180717"

//        如果是在本地windows执行需要加上这句代码
//        System.setProperty("hadoop.home.dir","D:/cdh/hadoop-2.6.0-cdh5.7.0");
        Configuration configuration = new Configuration();

//        写代码先写死，死去活来法
        FileSystem fileSystem = FileSystem.get(configuration);
        Path outputPath = new Path(output);
        if (fileSystem.exists(outputPath)){
            fileSystem.delete(outputPath,true);
        }

        Job job = Job.getInstance(configuration);
        job.setJarByClass(LogETLDriver.class);
        job.setMapperClass(LogETLMapper.class);
        job.setMapOutputKeyClass(NullWritable.class);
        job.setMapOutputValueClass(Text.class);
        FileInputFormat.setInputPaths(job,new Path(input));//"input/"
        FileOutputFormat.setOutputPath(job,new Path(output));//"output/d=20180717"
        job.waitForCompletion(true);

    }
}
  • 解释
    1
    2
    3
    4
    5
    6
    - 首先通过args读入两个参数作为输入路径和输出路径
    - 判断输出路径是否存在
      - 如果存在删除输出路径
    - 设置map类(MapperClass、MapOutputKeyClass、MapOutputValueClass)
    - 设置mapreduce的输入和输出路径
    - 设置结束后将程序设置为成功

代码打包并进行执行测试

  • 使用maven的package功能进行打包

  • 完毕后通过日志可以看到我们的jar包已经打出来了

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    [INFO] --- maven-jar-plugin:3.0.2:jar (default-jar) @ g6-hadoop ---
    [INFO] Building jar: D:\高铁6号\g6-train-hadoop\target\g6-hadoop-1.0.jar
    [INFO] ------------------------------------------------------------------------
    [INFO] BUILD SUCCESS
    [INFO] ------------------------------------------------------------------------
    [INFO] Total time: 16.362 s
    [INFO] Finished at: 2019-03-29T19:40:56+08:00
    [INFO] Final Memory: 30M/278M
    [INFO] ------------------------------------------------------------------------

    - jar目录在D:\高铁6号\g6-train-hadoop\target\g6-hadoop-1.0.jar

  • 将jar包上传到服务器上

    1
    2
    yum install lrzsz
    rz 我们的jar包

  • 将我们的原始日志文件上传到hdfs中

    1
    2
    3
    4
    baidu.log
    baidu	CN	A	E	[17/Jul/2018:17:07:50 +0800]	2	223.104.18.110	-	112.29.213.35:80	0	v2.go2yd.com	GET	http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c4b96b2226fd3f4c.mp4_bd.mp4	HTTP/1.1	-	bytes 13869056-13885439/25136186	TCP_HIT/206	112.29.213.35	video/mp4	17168	16384	-:0	0	11451601	-	"JSP3/2.0.14"	"-"	"-"	"-"	http	-	2	v1.go2yd.com	0.002	25136186	16384	-	-	-	-	-	-	-	-	1531818470104-11451601-112.29.213.66#2705261172	644514568
    baidu	CN	A	E	[17/Jul/2018:17:07:50 +0800]	2	223.104.18.110	-	112.29.213.35:80	0	v2.go2yd.com	GET	http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c4b96b2226fd3f4c.mp4_bd.mp4	HTTP/1.1	-	bytes 13869056-13885439/25136186	TCP_HIT/206	112.29.213.35	video/mp4	17168	16384	-:0	0	11451601	-	"JSP3/2.0.14"	"-"	"-"	"-"	http	-	2	v1.go2yd.com	0.002	25136186	16384	-	-	-	-	-	-	-	-	1531818470104-11451601-112.29.213.66#2705261172	644514568
    baidu	CN	A	E	[17/Jul/2018:17:07:50 +0800]	2	223.104.18.110	-	112.29.213.35:80	0	v2.go2yd.com	GET	http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c4b96b2226fd3f4c.mp4_bd.mp4	HTTP/1.1	-	bytes 13869056-13885439/25136186	TCP_HIT/206	112.29.213.35	video/mp4	17168	16384	-:0	0	11451601	-	"JSP3/2.0.14"	"-"	"-"	"-"	http	-	2	v1.go2yd.com	0.002	25136186	16384	-	-	-	-	-	-	-	-	1531818470104-11451601-112.29.213.66#2705261172	644514568
1
2
3
hadoop fs -mkdir /g6/logs
hadoop fs -put baidu.log /g6/logs
- 这里可以看到我们把日志文件放到hdfs的g6文件夹中了
  • 执行jar包的方法
    1
    hadoop jar g6-hadoop-1.0.jar com.ruozedata.hadoop.mapreduce.driver.LogETLDriver /g6/logs /g6/baidulog/day=20180717
1
2
3
4
5
6
7
8
9
我们在参数中指定了输出路径为/g6/baidulog/day=20180717
执行完mapreduce任务后，我们看一下hdfs输出的数据吧
hadoop fs -text /g6/baidulog/day=20180717/part-r-00000

baidu	CN	E	20180717170750	223.104.18.110	v2.go2yd.com	http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c4b96b2226fd3f4c.mp4_bd.mp4	16384	
baidu	CN	E	20180717170750	223.104.18.110	v2.go2yd.com	http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c4b96b2226fd3f4c.mp4_bd.mp4	16384	
baidu	CN	E	20180717170750	223.104.18.110	v2.go2yd.com	http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c4b96b2226fd3f4c.mp4_bd.mp4	16384	

可以看到我们的数据已经清洗出来了

结合hive

  • 新建一个外部表g6_access

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    create external table g6_access (
    cdn string,
    region string,
    level string,
    time string,
    ip string,
    domain string,
    url string,
    traffic bigint
    ) partitioned by (day string)
    ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
    LOCATION '/g6/baidulog/' ;

  • 通过alert命令刷新数据

    1
    2
    3
    4
    5
    6
    7
    8
    9
    alter table g6_access add if not exists partition(day='20180717');
    select * from g6_access;

    g6_access.cdn	g6_access.region	g6_access.level	g6_access.time	g6_access.ip	g6_access.domain	g6_access.url	g6_acc
    baidu	CN	E	20180717170750	223.104.18.110	v2.go2yd.com	http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c
    baidu	CN	E	20180717170750	223.104.18.110	v2.go2yd.com	http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c
    baidu	CN	E	20180717170750	223.104.18.110	v2.go2yd.com	http://v1.go2yd.com/user_upload/1531633977627104fdecdc68fe7a2c

    - 可以看到数据已经都进来了

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

大树，<br>人在首都，就职猪厂,<br>测试开发一枚<br><br>骐骥一跃不能十步，驽马十驾功在不舍<br><br>qq:565506019<br>